
package com.lee.lossycounting; 
 
import com.alibaba.fastjson.JSON; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.info.LkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.api.common.functions.RichFlatMapFunction; 
import org.apache.flink.api.common.state.MapState; 
import org.apache.flink.api.common.state.MapStateDescriptor; 
import org.apache.flink.configuration.Configuration; 
import org.apache.flink.util.Collector; 
 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.ArrayList; 
import java.util.List; 
 
public class AprioriGenFlowFunction implements FlatMapFunction<LkFlowInfo, CkFlowInfo> { 
    private int level;
    public AprioriGenFlowFunction(int level){
        this.level =  level;
    }
 
 
    @Override
    public void flatMap(LkFlowInfo lkFlowInfo, Collector<CkFlowInfo> collector) throws Exception {
        try {
 
            Logger.info(lkFlowInfo.getDIndex(),"/",lkFlowInfo.getDCIndex(),"LossyCounting AprioriGenFlowFunction -> lkFlowInfo ", lkFlowInfo);
            LocalDateTime start = LocalDateTime.now();
 
 
            List<List<String>> Ck = new ArrayList<>();
            List<String> Lk_1Flow = lkFlowInfo.getLkList();
            int k = level;
            long index = lkFlowInfo.getDIndex();
            long cindex = lkFlowInfo.getDCIndex();
            List<String> dataFlow = lkFlowInfo.getDataFlow();
 
            List<String> Lk_1i = Lk_1Flow;
            List<List<String>> LkFront = new ArrayList<>();
 
            if(Lk_1i.size() <= k - 2){
                return;
            }
 
            Lk_1i.sort(String::compareTo);
            dataFlow.sort(String::compareTo);
            String maxL = Lk_1i.get(Lk_1i.size() - 1);
            for(String item:dataFlow){
                if(item.compareTo(maxL) > 0){
                    List<String> newL = new ArrayList<>();
                    newL.addAll(Lk_1i);
                    newL.add(item);
                    LkFront.add(newL);
                }
            }
 
            LkFront.stream().forEach(Lk_1j -> {
                List<String> Ckitem = Lk_1j;
                Logger.debug(index, "BdsAprioriGenFlowFunction -> Ckitem ", Ckitem, " dataFlow ", dataFlow);
                if (!ArrayUtils.isListItemInArray(Ckitem, Ck)) {
                    Ck.add(Ckitem);
                    CkFlowInfo ckFlowInfo = new CkFlowInfo();
                    ckFlowInfo.setDIndex(index);
                    ckFlowInfo.setDCIndex(cindex);
                    ckFlowInfo.setDataFlow(dataFlow);
                    ckFlowInfo.setCkList(Ckitem);
                    collector.collect(ckFlowInfo);
                }
            });
            DMapUtils.addFinishIndex(FinishQualifierEnum.APRIORI_GEN.getCode(),1l);
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(start, end);
            Logger.info(lkFlowInfo.getDIndex(),"/",lkFlowInfo.getDCIndex(),between.toMillis(),"END LossyCounting AprioriGenFlowFunction -> lkFlowInfo ", lkFlowInfo);
        }catch(Exception e){
            //释放锁
            String lockKey = JSON.toJSONString(lkFlowInfo.getLkList());
 
            Logger.error("LossyCounting BdsAprioriGenFlowFunction -> 置信度计算报错：",e);
            e.printStackTrace();
        }
 
    }
} 
